package manager

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"sieve_engine/execute"
	_ "sieve_engine/execute/facebook"
	_ "sieve_engine/execute/instgram"
	_ "sieve_engine/execute/linkedin"
	_ "sieve_engine/execute/twitter"
	_ "sieve_engine/execute/whatsapp"
	"sieve_engine/lib/log"
	"sieve_engine/task"
	"sieve_engine/utils"

	"github.com/RoaringBitmap/roaring/roaring64"
	"github.com/go-redis/redis/v8"
)

type TaskExecutorService struct {
	locker        sync.Mutex
	taskInfo      task.Info
	progress      task.Result // 总进度
	redisCli      *redis.Client
	progressChan  chan *task.Result // 每次接收到的进度
	counter       *Counter
	dispatchCount *atomic.Int32
}

func writeProgress(s interface{}, result *task.Result) {

	service := s.(*TaskExecutorService)
	result.Progress.State = service.progress.State
	service.writeProgress(result)
	return
}

func newTaskExecutorService(taskInfo task.Info, manager *TaskManagerService) *TaskExecutorService {

	return &TaskExecutorService{
		progress: task.Result{
			Progress: taskInfo.Progress,
		},
		taskInfo:     taskInfo,
		redisCli:     manager.redisCli,
		progressChan: make(chan *task.Result, 10240),
		counter:      NewCounter(60 * 60 * 2), // 两个小时
	}
}

func (s *TaskExecutorService) AddDispatchCount(delta int32) int32 {

	if s.dispatchCount == nil {
		s.dispatchCount = &atomic.Int32{}
	}
	return s.dispatchCount.Add(delta)
}

func (s *TaskExecutorService) GetDispatchCount() *int32 {

	if s.dispatchCount == nil {
		return nil
	}
	data := s.dispatchCount.Load()
	return &data
}

func (s *TaskExecutorService) stats() (string, error) {

	s.locker.Lock()
	defer s.locker.Unlock()
	b, err := json.Marshal(s.progress)
	return string(b), err
}

func (s *TaskExecutorService) readProgress() chan *task.Result {
	return s.progressChan
}

func (s *TaskExecutorService) writeProgress(progress *task.Result) {

	s.locker.Lock()
	s.progress.Increase(progress)
	s.taskInfo.FinishCount = s.progress.FinishCount
	s.progressChan <- progress
	if s.progress.FinishCount == int64(s.taskInfo.TaskCount) {
		close(s.progressChan)
	}
	s.locker.Unlock()
	s.counter.Inc(MonitorSample{Delta: float64(progress.FinishCount)})
	return
}

func (s *TaskExecutorService) start() {

	s.progress.State = task.StatusRun
	s.taskInfo.UpdateAt = time.Now()
	return
}

func (s *TaskExecutorService) pause() {

	s.progress.State = task.StatusPause
	s.taskInfo.UpdateAt = time.Now()
	return
}

func (s *TaskExecutorService) terminate() {

	s.progress.State = task.StatusTerminate
	t := time.Now()
	s.taskInfo.UpdateAt = t
	s.taskInfo.ElapsedTime = t.Sub(s.taskInfo.CreateAt).String()
	close(s.progressChan)
	return
}

func (s *TaskExecutorService) finish() {

	s.progress.State = task.StatusFinish
	t := time.Now()
	s.taskInfo.UpdateAt = t
	s.taskInfo.ElapsedTime = t.Sub(s.taskInfo.CreateAt).String()

	_ = s.dumpResultToFile()
	return
}

func (s *TaskExecutorService) clearResult() {

	s.progress.Clear()
	return
}

func (s *TaskExecutorService) checkPauseState(manager *TaskManagerService) bool {

	if s.progress.State == task.StatusPause {
		return true
	}

	return false
}

func (s *TaskExecutorService) checkTerminateState(manager *TaskManagerService) bool {

	if s.progress.State == task.StatusTerminate {
		return true
	}
	return false
}

func (s *TaskExecutorService) checkFinishState(manager *TaskManagerService) bool {

	// 任务最终落地写入到redis，状态才会变为 StatusFinish，筛料任务任务计数就会达标
	if s.progress.State == task.StatusFinish && s.progress.FinishCount == int64(s.taskInfo.TaskCount) {
		return true
	}

	return false
}

func (s *TaskExecutorService) checkWaitingState(manager *TaskManagerService) bool {

	/* 收到 暂停指令，那么就不接受 redis 的消息筛料，处于停止状态，由redis保存现有的状态 */
	if s.progress.State == task.StatusPause {
		return true
	}

	if s.progress.State == task.StatusTerminate {
		return true
	}

	return false
}

func (s *TaskExecutorService) dumpResultToFile() error {

	var (
		ctx       = context.Background()
		phoneKey  = fmt.Sprintf("%v", s.taskInfo.ResultListKey)
		batchSize = 500
		bitmap    = roaring64.NewBitmap()
	)

	for {
		results, err := manager.redisCli.RPopCount(ctx, phoneKey, batchSize).Result()
		if err != nil {
			if err == redis.Nil {
				log.Info(fmt.Sprintf("任务ID:%v dump Redis 结果 从 REDIS(%v) 消耗 (筛源:%v,实际:%v) 错误:%+v 退出循环",
					s.taskInfo.Id, phoneKey, batchSize, len(results), err))
				break
			}
			log.Err(fmt.Sprintf("任务ID:%v 从 REDIS(%v) 消耗 (筛源:%v,实际:%v) 错误:%+v", s.taskInfo.Id, phoneKey, batchSize, len(results), err))
			break
		}

		for _, result := range results {

			if strings.Contains(result, ">in") {

				n := result[0 : len(result)-len(">in")]
				if x, err := strconv.ParseUint(n, 10, 64); err != nil {
					log.Err(fmt.Sprintf("号码(%v)错误:%v", result, err))
				} else {
					bitmap.Add(x)
				}
			}

			// if x, err := strconv.ParseUint(result, 10, 64); err != nil {
			//	log.Err(fmt.Sprintf("号码(%v)错误:%v", result, err))
			// } else {
			//	bitmap.Add(x)
			// }
		}

	}

	if err := utils.ToFileFromBitMap(getDumpFilePath(s.taskInfo.Id, fileTypeResult), bitmap, false); err != nil {
		return err
	}

	return nil
}

func (s *TaskExecutorService) run(executor execute.TaskExecutorInterface, manager *TaskManagerService) error {

	var (
		cursor        = uint64(0)
		batchSize     = int64(manager.taskConfig.fetchCountOnTime)
		ctx           = context.Background()
		cpg           = manager.taskConfig.maxGorCountPerTask // count per task 每个筛料最大指派 cpg 个协程并发跑
		cpgCounter    = &atomic.Int32{}
		bitmap        = roaring64.NewBitmap()
		phoneKey      = fmt.Sprintf("%v", s.taskInfo.SrcListKey)
		srcCount      int64
		allResults    []string
		lastPrintTime = time.Now().Unix()
		materialSize  = int64(1000)
	)

	if _count, err := s.redisCli.SCard(ctx, phoneKey).Result(); err != nil {
		log.Err(fmt.Sprintf("任务ID:%v 从获取数量失败:%v", s.taskInfo.Id, err))
	} else {
		srcCount = _count
	}

	for {
		results, newCursor, err := s.redisCli.SScan(ctx, phoneKey, cursor, "", materialSize).Result()
		if err != nil {
			if err == redis.Nil {
				log.Info(fmt.Sprintf("任务ID:%v 从 REDIS(%v) 消耗 (筛源:%v,实际:%v) 错误:%+v 退出循环",
					s.taskInfo.Id, phoneKey, materialSize, len(results), err))
				return err
			}
			log.Err(fmt.Sprintf("任务ID:%v 从 REDIS(%v) 消耗 (筛源:%v,实际:%v) 错误:%+v", s.taskInfo.Id, phoneKey, materialSize, len(results), err))
			return err
		} else {

			cursor = newCursor
			for _, v := range results {
				if x, err := strconv.ParseUint(v, 10, 64); err == nil {
					if bitmap.Contains(x) {
						continue
					}

					bitmap.Add(x)

					allResults = append(allResults, v)

				}
			}
		}

		now := time.Now().Unix()
		if now > lastPrintTime+10 {
			lastPrintTime = now
			log.Info(fmt.Sprintf("任务ID：%v 已经读取任务数:%v 总任务数:%+v", phoneKey, len(allResults), srcCount))
		}
		if bitmap.GetCardinality() >= uint64(srcCount) {
			break
		}
	}

	log.Info(fmt.Sprintf("读取到任务:%v 料子总数:%v", s.taskInfo.Id, bitmap.GetCardinality()))
	spliceSlice := utils.ChunkSliceString(allResults, int(batchSize))
	for i := 0; i < len(spliceSlice); i++ {

		/* 收到 暂停指令，那么就不接受 redis 的消息筛料，处于停止状态，由redis保存现有的状态 */
		if s.checkWaitingState(manager) {
			i = i - 1
			time.Sleep(time.Millisecond * 5)
			continue
		}

		batchIndex := i
		results := spliceSlice[batchIndex]
		gorCounts := int32(len(results))
		// runningGors := manager.taskConfig.runningGors.Load()
		// if runningGors+gorCounts > manager.taskConfig.maxGorCount {
		//	now := time.Now().Unix()
		//	if now > lastPrintTime+10 {
		//		lastPrintTime = now
		//		log.Info(fmt.Sprintf("超过系统最大可运行协程数(运行中协程数量为:%v 当前筛料任务需要协程数:%v,系统最大可同时运行协程数:%+v)", runningGors, gorCounts, manager.taskConfig.maxGorCount))
		//	}
		//	i = i - 1
		//	time.Sleep(time.Millisecond * 5)
		//	continue
		// }

		gpgValue := cpgCounter.Load()
		// if gpgValue+gorCounts > cpg {
		//	now := time.Now().Unix()
		//	if now > lastPrintTime+10 {
		//		lastPrintTime = now
		//		log.Info(fmt.Sprintf("超过每个任务最大可运行协程数(运行中协程数量为:%v 当前筛料任务需要协程数:%v,每个任务最大可同时运行协程数:%+v)", runningGors, gorCounts, manager.taskConfig.maxGorCountPerTask))
		//	}
		//	i = i - 1
		//	time.Sleep(time.Millisecond * 5)
		//	continue
		// }
		//
		consumeStart := time.Now()
		cpgCounter.Add(gorCounts)
		manager.taskConfig.runningGors.Add(gorCounts)
		now := time.Now().Unix()
		if now > lastPrintTime+10 {
			lastPrintTime = now
			log.Info(fmt.Sprintf("任务(ID:%v,当前批次号:%v) 协程信息:(运行中:%v 实际:%v),从 REDIS(%v) 消耗 (筛源:%v,实际:%v) 数据耗时(%v)",
				s.taskInfo.Id, batchIndex,
				gpgValue,
				cpg,
				phoneKey, batchSize, len(results),
				time.Since(consumeStart).String()))
		}
		/*
			s.progress 用作局部小量任务缓存使用
		*/

		var (
			tt        = time.Now().Unix()
			fNotifyCb = func(result *task.Result) bool {

				var (
					t = time.Now().Unix()
				)

				// log.Info(fmt.Sprintf("%p %v %v %v %v %v %v \n", result, result.FinishCount, result.FinishCount >= 200, t-tt >= 8, result.AllFinishCount, result.Finished(), len(result.Phones())))
				if result.FinishCount >= 200 || t-tt >= 10 || result.Finished() {
					tt = t
					return true
				}

				return false
			}
			progress = task.NewTaskResult(s, results, writeProgress, fNotifyCb).WithTaskState(task.StatusRun)
			params   = execute.TaskExecutorParams{
				BatchIndex:  batchIndex,
				GorCount:    len(results),
				TaskInfo:    &s.taskInfo,
				GorCounter:  cpgCounter,
				RunningGors: &manager.taskConfig.runningGors,
			}
		)

		// newValue := s.AddDispatchCount(gorCounts)
		_ = s.AddDispatchCount(gorCounts)
		// log.Info(fmt.Sprintf("任务ID:%v dispatch new1:%+v", s.taskInfo.Id, newValue))
		if err := executor.Execute(params, progress); err != nil {
			log.Err(fmt.Sprintf("任务ID:%v 执行筛料失败 %+v", s.taskInfo.Id, err))
		}
	}

	return nil
}
